package com.demo.java.tsdb;

import com.clearspring.analytics.util.Lists;
import com.demo.java.utils.JSONUtil;
import com.stumbleupon.async.Callback;
import com.stumbleupon.async.Deferred;
import lombok.extern.slf4j.Slf4j;
import net.opentsdb.core.*;
import net.opentsdb.query.expression.EDPtoDPS;
import net.opentsdb.query.expression.ExpressionFactory;
import net.opentsdb.utils.Config;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

@Slf4j
public class OpenTsdbTest {

    private static Config config;

    public void query(String json) throws Exception {
        config = new Config();
        TSDB tsdb = new TSDB(config);

        TSQuery tsQuery = JSONUtil.toObject(json, TSQuery.class);
        Query[] queries = tsQuery.buildQueries(tsdb);

        int nqueries = queries.length;
        // 异步执行
        List<Deferred<DataPoints[]>> dataPointsCB = Lists.newArrayList();
        for (Query query : queries) {
            dataPointsCB.add(query.runAsync());
        }
        Deferred.groupInOrder(dataPointsCB)
                .addCallback(new QueriesCB(nqueries, tsQuery))
                .addErrback(new QueriesEB())
                .addCallback(new AlignCB())
                .addErrback(new QueriesEB())
                .join();
    }

    static class QueriesCB implements Callback<ArrayList<DataPoints[]>, ArrayList<DataPoints[]>> {
        private int nqueries;
        private TSQuery query;

        public QueriesCB(int nqueries, TSQuery query) {
            this.nqueries = nqueries;
            this.query = query;
        }

        public ArrayList<DataPoints[]> call(ArrayList<DataPoints[]> queryResults) throws Exception {
            CalcSeries expr = (CalcSeries) ExpressionFactory.getByName("calcSeries");
            ArrayList<DataPoints[]> results = new ArrayList(nqueries);
            ArrayList<DataPoints[]> result = new ArrayList();
            ArrayList<String> param = new ArrayList();
            for (DataPoints[] dataPoints : queryResults) {
                if (dataPoints.length > 1) {
                    for (DataPoints dprsult : dataPoints) {
                        result.add(new DataPoints[]{ dprsult });
                    }
                    continue;
                }
                result.add(dataPoints);
            }
            ArrayList<DataPoints[]> expression_results = new ArrayList();
            expression_results.add(expr.evaluate(query, result, param));
            results.addAll(expression_results);
            return results;
        }
    }

    static class QueriesEB implements Callback<Object, Exception> {
        public Object call(Exception e) throws Exception {
            log.error("queries failed", e);
            return null;
        }
    }

    static class AlignCB implements Callback<Qianfan, ArrayList<DataPoints[]>> {
        public Qianfan call(ArrayList<DataPoints[]> dataPoints) throws Exception {
            DataPoints[] dataSets = dataPoints.get(0);
            DataPoints data = null;
            if (dataSets.length > 0) {
                data = dataSets[0];
                // Collection<TSSubQuery> values = ((EDPtoDPS)data).getIterator().getTsSubQueries().values();
                // List<TSSubQuery> queries = query.getQueries();
            }
            return null;
        }
    }
}
